package com.galeno.day09;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2610:15
 * FLink ProcessFunction,Fink底层的算子,可以访问flink底层的属性和方法
 * 1.处理数据 element
 * 2.使用状态,只能使用keyedStream
 * 3.使用定时器,只能使用keyedStream
 */
public class TimeProcessFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> lines = env.socketTextStream("192.168.77.3", 9999);
        //spark,1
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAnd1 = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] split = value.split("\\s+");
                return Tuple2.of(split[0], Integer.parseInt(split[1]));
            }
        });
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAnd1.keyBy(x -> x.f0);
        keyedStream.process(new KeyedProcessFunction<String, Tuple2<String, Integer>, Tuple2<String,Integer>>() {

            @Override
            public void open(Configuration parameters) throws Exception {

            }

            @Override
            public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                //processElement中不输出数据,而是将数据攒起来,然后注册定时器
                //攒起来的数据放在状态中(可以容错)

                ctx.timerService().registerProcessingTimeTimer(System.currentTimeMillis()+15000);


            }
            //每个key都要自己的定时器
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {
                System.out.println("触发定时器的时间"+timestamp);
            }
        });


    }
}
